package org.code.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;

/**
 * @author xiaojian
 * @classname HdfsClient
 * @date 2023/12/3 16:23
 * @description HDFS
 */
public class HdfsClient {

    public static final String user = "xiao";

    public static final String hdfs_uri = "hdfs://192.168.67.102:8020";

    private FileSystem fs;

    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        URI uri = new URI(hdfs_uri);
        Configuration configuration = new Configuration();
        fs = FileSystem.get(uri, configuration, user);
    }

    @After
    public void close() throws IOException {
        fs.close();
    }

    /**
     * 创建目录
     *
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    @Test
    public void mkdir() throws URISyntaxException, IOException, InterruptedException {
        // 创建目录
        fs.mkdirs(new Path("/xiyou/huaguoshan"));
    }

    /**
     * 上传文件
     *
     * @throws IOException
     */
    @Test
    public void copyFromLocalFile() throws IOException {
        /**
         * delSrc：是否删除源文件
         * overwrite：是否覆盖hdfs目标文件
         * src：源文件
         * dst：目标文件
         */
        fs.copyFromLocalFile(false, true,
                new Path("/Users/xiaojian/Desktop/hadoop/sunwukong.txt"),
                new Path("/xiyou/huaguoshan"));
    }

    /**
     * 文件下载
     *
     * @throws IOException
     */
    @Test
    public void copyToLocalFile() throws IOException {
        // boolean delSrc 指是否将原文件删除
        // Path src 指要下载的文件路径
        // Path dst 指将文件下载到的路径
        // boolean useRawLocalFileSystem 是否开启文件校验
//        fs.copyToLocalFile(false,
//                new Path("/xiyou/huaguoshan/sunwukong.txt"),
//                new Path("/Users/xiaojian/Desktop/hadoop"),
//                true);

        // crc-32
        fs.copyToLocalFile(false,
                new Path("/a.txt"),
                new Path("/Users/xiaojian/Desktop/hadoop"),
                false);
    }

    /**
     * 重命名
     *
     * @throws IOException
     */
    @Test
    public void rename() throws IOException {
        fs.rename(new Path("/xiyou/huaguoshan/sunwukong.txt"),
                new Path("/xiyou/huaguoshan/sunwukong1.txt"));
    }

    /**
     * 删除文件
     *
     * @throws IOException
     */
    @Test
    public void delete() throws IOException {
        /**
         * recursive: 是否递归删除
         */
        fs.delete(new Path("/xiyou"), true);
    }

    /**
     * 获取文件详情
     *
     * @throws IOException
     */
    @Test
    public void listFile() throws IOException {
        // 获取文件详情
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);

        while (listFiles.hasNext()) {
            LocatedFileStatus fileStatus = listFiles.next();
            // 获取文件路径
            System.out.printf("============ %s ============\n", fileStatus.getPath());
            // 获取文件权限
            System.out.printf("permission: %s \n", fileStatus.getPermission());
            // 获取文件所有者
            System.out.printf("owner: %s \n", fileStatus.getOwner());
            // 获取文件所属组
            System.out.printf("group: %s\n", fileStatus.getGroup());
            // 获取文件大小
            System.out.printf("len: %s\n", fileStatus.getLen());
            // 获取文件最后修改时间
            System.out.printf("ModificationTime: %s\n", fileStatus.getModificationTime());
            // 获取文件副本
            System.out.printf("replication: %s\n", fileStatus.getReplication());
            // 获取块大小
            System.out.printf("BlockSize: %s\n", fileStatus.getBlockSize());
            // 获取文件名称
            System.out.printf("fileName: %s\n", fileStatus.getPath().getName());
            // 获取块信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            System.out.printf("blockLocations: %s\n", Arrays.toString(blockLocations));
        }
    }

    /**
     * 文件和文件夹判断
     *
     * @throws IOException
     */
    @Test
    public void listStatus() throws IOException {
        FileStatus[] listStatus = fs.listStatus(new Path("/"));

        for (FileStatus status : listStatus) {
            String name = status.getPath().getName();
            if (status.isFile()) {
                System.out.printf("f: %s\n", name);
            } else {
                System.out.printf("d: %s\n", name);
            }
        }
    }
}
